package com.hhf.rrd.usage.sink;

import com.hhf.rrd.model.Student;
import com.hhf.rrd.utils.MysqlUtils;
import com.sun.xml.internal.fastinfoset.Encoder;
import org.apache.flink.api.common.functions.IterationRuntimeContext;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.serialization.SimpleStringEncoder;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;

import java.sql.Connection;
import java.sql.PreparedStatement;

/**
 * sink
 *
 * @author huanghaifeng15
 * @date 2022/2/10 09:09
 **/
public class SinkApp {

    public static void main(String[] args) throws Exception {
        // 获取运行环境上下文
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 获取流式数据源
        DataStreamSource<String> source = env.readTextFile("data/user_register.log");

        // 处理词频的业务逻辑  Transformation
        SingleOutputStreamOperator<Student> streamOperator = source.map(new MapFunction<String, Student>() {
            @Override
            public Student map(String value) throws Exception {
                Student student = new Student();
                String[] studentArr = value.split(",");
                student.setName(studentArr[0]);
                student.setAge(Integer.valueOf(studentArr[1]));
                student.setSex(Integer.valueOf(studentArr[2]));
                return student;
            }
        }).keyBy(new KeySelector<Student, String>() {
            @Override
            public String getKey(Student value) throws Exception {
                return value.getName();
            }
        }).sum("sex");

        streamOperator.print();

        // to-->redis sink算子
        FlinkJedisPoolConfig jedisPoolConfig = new FlinkJedisPoolConfig.Builder()
                .setHost("127.0.0.1").build();

        streamOperator.map(new MapFunction<Student, Tuple2<String,Integer>>() {
            @Override
            public Tuple2<String, Integer> map(Student value) throws Exception {
                return Tuple2.of(value.getName(), value.getSex());
            }
        }).addSink(new RedisSink<>(jedisPoolConfig, new RedisMapper<Tuple2<String, Integer>>() {
            @Override
            public RedisCommandDescription getCommandDescription() {
                return new RedisCommandDescription(RedisCommand.HSET, "hhf-user");
            }

            @Override
            public String getKeyFromData(Tuple2<String, Integer> data) {
                return data.f0;
            }

            @Override
            public String getValueFromData(Tuple2<String, Integer> data) {
                return String.valueOf(data.f1);
            }
        }));

        // to-->mysql sink算子
        streamOperator.addSink(new RichSinkFunction<Student>() {
            private Connection connection;
            private PreparedStatement insertStatement;
            private PreparedStatement updateStatement;

            @Override
            public void setRuntimeContext(RuntimeContext t) {
                super.setRuntimeContext(t);
            }

            @Override
            public RuntimeContext getRuntimeContext() {
                return super.getRuntimeContext();
            }

            @Override
            public IterationRuntimeContext getIterationRuntimeContext() {
                return super.getIterationRuntimeContext();
            }

            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);
                connection = MysqlUtils.getConnection();
                assert connection != null;
                insertStatement = connection.prepareStatement("insert into student(name,sex) values (?,?)");
                updateStatement = connection.prepareStatement("update student set sex=? where name=?");
            }

            @Override
            public void close() throws Exception {
                super.close();
                MysqlUtils.close(connection,insertStatement);
                MysqlUtils.close(connection,updateStatement);
            }

            /**
             * 有一条数据，执行一次
             */
            @Override
            public void invoke(Student value, Context context) throws Exception {
                super.invoke(value, context);
                System.out.println("=====invoke======" + value.getName() + "-->" + value.getSex());

                updateStatement.setString(1, value.getName());
                updateStatement.setInt(2 , value.getSex());
                updateStatement.execute();

                if(updateStatement.getUpdateCount() == 0) {
                    insertStatement.setString(1, value.getName());
                    insertStatement.setInt(2 , value.getSex());
                    insertStatement.execute();
                }
            }
        });

        streamOperator.writeAsText("out", FileSystem.WriteMode.OVERWRITE);
        // streamOperator.addSink(StreamingFileSink.forRowFormat(new Path("out"), new SimpleStringEncoder()));

        // 执行任务
        env.execute("sinkApp");
    }
}
